/**
 * Copyright 2009 Sergio Bossa
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package com.googlecode.actorom.remote;

import com.googlecode.actorom.Actor;
import com.googlecode.actorom.Address;
import com.googlecode.actorom.Topology;
import com.googlecode.actorom.impl.remote.RemoteActorProxy;
import com.googlecode.actorom.impl.remote.channel.RemoteChannelConnector;
import com.googlecode.actorom.impl.remote.channel.ClientCallback;
import com.googlecode.actorom.impl.remote.channel.protocol.Header;
import com.googlecode.actorom.impl.remote.channel.protocol.CompleteFutureCommand;
import com.googlecode.actorom.impl.remote.channel.protocol.PingCommand;
import com.googlecode.actorom.impl.remote.channel.protocol.SpawnCommand;
import com.googlecode.actorom.impl.remote.channel.protocol.support.ProtocolUtils;
import com.googlecode.actorom.support.ActorCreationException;
import com.googlecode.actorom.support.TopologyException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import net.jcip.annotations.ThreadSafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * {@link com.googlecode.actorom.Topology} implementation which remotely connects to a given {@link ServerTopology} and uses
 * its services and actors.<br>
 * A client topology is able to:
 * <ul>
 * <li>Spawn actors into the remote server topology.</li>
 * <li>Access actors from the remote server topology.</li>
 * </ul>
 * In order to do so, all objects exchanged with the remote topology, that is all handlers of remotely spawned actors, as well as all
 * messages sent to remote actors, must be serializable.
 * <br><br>
 * Once connected, the client topology will send ping messages to the server in order to keep alive the connection: 
 * if no response is received in a given amount of time, the client will be forcefully disconnected after a given number of failed retries.
 *
 * @author Sergio Bossa
 */
@ThreadSafe
public class ClientTopology implements Topology {

    private static final transient Logger LOG = LoggerFactory.getLogger(ClientTopology.class);
    private final String host;
    private final int port;
    private final long communicationTimeout;
    private final TimeUnit timeoutUnit;
    private final long keepAlive;
    private final TimeUnit keepAliveUnit;
    private final int keepAliveRetries;
    private final Map<Address, RemoteActorProxy> remoteActors;
    private final RemoteChannelConnector topologyConnector;
    private final ScheduledExecutorService connectionKeepAliveScheduler;
    private boolean connected;
    private boolean shutdown;

    /**
     * Create a client topology and connect to a remote server topology (see {@link ServerTopology}).
     *
     * @param host The host of the server topology to connect to.
     * @param port The port of the server topology to connect to.
     * @param communicationTimeout Timeout for any kind of communication between client and server.
     * @param timeoutUnit Timeout unit.
     * @param keepAlive Time between keep-alive ping messages sent from client to server: client will be disconnected
     * after the first failed exchange.
     * @param keepAliveUnit Keep-alive time unit.
     */
    public ClientTopology(String host, int port, long communicationTimeout, TimeUnit timeoutUnit, long keepAlive, TimeUnit keepAliveUnit) {
        this(host, port, communicationTimeout, timeoutUnit, keepAlive, keepAliveUnit, 1);
    }

    /**
     * Create a client topology and connect to a remote server topology (see {@link ServerTopology}).
     *
     * @param host The host of the server topology to connect to.
     * @param port The port of the server topology to connect to.
     * @param communicationTimeout Timeout for any kind of communication between client and server.
     * @param timeoutUnit Timeout unit.
     * @param keepAlive Time between keep-alive ping messages sent from client to server.
     * @param keepAliveUnit Keep-alive time unit.
     * @param keepAliveRetries Number of keep-alive failed retries before disconnecting the client.
     */
    public ClientTopology(String host, int port, long communicationTimeout, TimeUnit timeoutUnit, long keepAlive, TimeUnit keepAliveUnit, int keepAliveRetries) {
        this.host = host;
        this.port = port;
        this.communicationTimeout = communicationTimeout;
        this.timeoutUnit = timeoutUnit;
        this.keepAlive = keepAlive;
        this.keepAliveUnit = keepAliveUnit;
        this.keepAliveRetries = keepAliveRetries;
        this.remoteActors = new HashMap<Address, RemoteActorProxy>();
        // Connect to remote server topology:
        this.topologyConnector = new RemoteChannelConnector(host, port, TimeUnit.MILLISECONDS.convert(communicationTimeout, timeoutUnit), new ClientTopologyCallback());
        this.topologyConnector.connect();
        // Start the connection checker:
        this.connectionKeepAliveScheduler = Executors.newScheduledThreadPool(1);
        this.connectionKeepAliveScheduler.scheduleWithFixedDelay(new ConnectionKeepAlive(), keepAlive, keepAlive, keepAliveUnit);
        //
        this.connected = true;
        this.shutdown = false;
    }

    public final synchronized Address spawnActor(String id, Object handler) {
        if (connected && !shutdown) {
            Address spawnedAddress = doSpawnActor(id, handler);
            if (spawnedAddress != null) {
                return spawnedAddress;
            } else {
                return null;
            }
        } else {
            throw new TopologyException("Topology is disconnected or shutdown!");
        }
    }

    public final synchronized Actor spawnAndGetActor(String id, Object handler) {
        if (connected && !shutdown) {
            Address spawnedAddress = doSpawnActor(id, handler);
            if (spawnedAddress != null) {
                return doGetActor(spawnedAddress);
            } else {
                return null;
            }
        } else {
            throw new TopologyException("Topology is disconnected or shutdown!");
        }
    }

    public final synchronized Actor getActor(Address address) {
        if (connected && !shutdown) {
            if (address.isRemote() && address.getHost().equals(host) && address.getPort() == port) {
                return doGetActor(address);
            } else {
                throw new TopologyException("Not a remote address for server topology: " + address);
            }
        } else {
            throw new TopologyException("Topology is disconnected or shutdown!");
        }
    }

    public final synchronized void shutdown() {
        if (!shutdown) {
            shutdown = true;
            connected = false;
            connectionKeepAliveScheduler.shutdownNow();
            topologyConnector.disconnect();
            doDisconnectRemoteActors();
        }
    }

    public final synchronized boolean isConnected() {
        return connected;
    }

    private Address doSpawnActor(String id, Object handler) {
        Header header = new Header(ProtocolUtils.generateCorrelationId());
        boolean acknowledged = topologyConnector.send(new SpawnCommand(header, id, handler));
        if (acknowledged) {
            return Address.newRemoteAddress(host, port, id);
        } else {
            throw new ActorCreationException("Cannot spawn remote actor with id " + id + " on " + host + ":" + port);
        }
    }

    private Actor doGetActor(Address address) {
        RemoteActorProxy actor = remoteActors.get(address);
        if (actor != null && actor.isActive()) {
            return actor;
        } else if (actor == null) {
            RemoteChannelConnector actorConnector = new RemoteChannelConnector(host, port, TimeUnit.MILLISECONDS.convert(communicationTimeout, timeoutUnit), new ClientActorCallback());
            actor = new RemoteActorProxy(address, actorConnector);
            if (actor.isActive()) {
                remoteActors.put(address, actor);
                return actor;
            } else {
                return null;
            }
        } else {
            actor.disconnect();
            remoteActors.remove(address);
            return null;
        }
    }

    private void doDisconnectRemoteActors() {
        Iterator<RemoteActorProxy> actors = remoteActors.values().iterator();
        while (actors.hasNext()) {
            RemoteActorProxy actor = actors.next();
            actor.disconnect();
            actors.remove();
        }
    }

    private class ClientActorCallback implements ClientCallback {

        public void completeFuture(CompleteFutureCommand command) {
            synchronized (ClientTopology.this) {
                Address address = command.getRecipient();
                RemoteActorProxy actor = remoteActors.get(address);
                if (actor != null) {
                    actor.completeFuture(command.getHeader().getCorrelationId(), command.getResult(), command.getException());
                }
            }
        }
    }

    private class ClientTopologyCallback implements ClientCallback {

        public void completeFuture(CompleteFutureCommand command) {
            throw new UnsupportedOperationException();
        }
    }

    private class ConnectionKeepAlive implements Runnable {

        private int failedRetries = 0;

        public void run() {
            synchronized (ClientTopology.this) {
                if (connected && !shutdown) {
                    try {
                        LOG.debug("Checking connection to " + host + ":" + port);
                        boolean acknowledged = topologyConnector.send(new PingCommand(new Header(ProtocolUtils.generateCorrelationId())));
                        if (acknowledged) {
                            failedRetries = 0;
                            Iterator<RemoteActorProxy> actors = remoteActors.values().iterator();
                            while (actors.hasNext()) {
                                RemoteActorProxy actor = actors.next();
                                if (!actor.isActive()) {
                                    LOG.warn("Actor at address " + actor.getAddress() + " seems to be no more active, removing.");
                                    actor.disconnect();
                                    actors.remove();
                                }
                            }
                        } else {
                            failedRetries++;
                            if (failedRetries == keepAliveRetries) {
                                connected = false;
                                LOG.warn("Connection check to " + host + ":" + port + " failed!");
                            }
                        }
                    } catch (Exception ex) {
                        LOG.warn(ex.getMessage(), ex);
                    }
                }
            }
        }
    }
}
